package com.sh.data.engine.domain.datadev.flink.job;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.sh.data.engine.common.enumDefinition.OfflineSyncTaskStatus;
import com.sh.data.engine.domain.integration.online.service.FlinkDataSyncService;
import com.sh.data.engine.infrastructure.config.FlinkConfig;
import com.sh.data.engine.job.core.util.CmdUtil;
import com.sh.data.engine.repository.dao.integration.online.entity.FlinkDataSyncTaskEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Component
@Slf4j
public class CheckFlinkStatusTask {

    @Autowired
    private FlinkConfig flinkConfig;

    @Autowired
    private FlinkDataSyncService flinkDataSyncService;

    /** 主要目的：定时检查flink是否启动，挂了重新拉起 */
    @Scheduled(cron = "0 0/5 * * * ?")
    public void checkFlinkStatus() {

        String flinkHome = flinkConfig.getHome();
        String port = flinkConfig.getPort();
        String cmd = flinkHome + "start-cluster.sh";
        try {
            // FIXME 只检测JobManager服务器节点
            Socket localhost = new Socket("localhost", Integer.parseInt(port));
        } catch (IOException e) {
            try {
                // runCmd("root",args.toArray(new String[args.size()]));
                CmdUtil.runCmd(cmd);
            } catch (Exception exception) {
                log.info("启动flink服务失败");
            }
        }

        String[] result = new String[40];
        String listCmd = flinkHome + "flink list";
        try {
            result = CmdUtil.runCmd(listCmd);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        log.info("checkFlinkStatus result" + result.toString());
        List<String> list1 = new ArrayList<>();
        if (result != null && result.length > 0) {
            String pattern = "\\b[0-9a-f]{32}\\b";
            Pattern r = Pattern.compile(pattern);
            for (String line : result) {
                Matcher m = r.matcher(line);
                while (m.find()) {
                    list1.add(m.group().trim());
                }
            }
        }
        log.info("checkFlinkStatus" + list1.toString());
        List<FlinkDataSyncTaskEntity> list =
                flinkDataSyncService.list(
                        new LambdaQueryWrapper<>(FlinkDataSyncTaskEntity.class)
                                .eq(FlinkDataSyncTaskEntity::getTaskStatus, 2)
                                .isNotNull(FlinkDataSyncTaskEntity::getFlinkJobId));
        for (FlinkDataSyncTaskEntity flinkDataSyncTaskEntity : list) {
            String flinkJobId = flinkDataSyncTaskEntity.getFlinkJobId();
            if (list1.contains(flinkJobId.trim())) {
                flinkDataSyncTaskEntity.setTaskRunStatus(OfflineSyncTaskStatus.running.getTaskStatus());
            } else {
                flinkDataSyncTaskEntity.setTaskRunStatus(OfflineSyncTaskStatus.failure.getTaskStatus());
            }
        }
        flinkDataSyncService.updateBatchById(list);
    }
}
